package homework

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo4Work {

    def main(args: Array[String]): Unit = {
      val spark: SparkSession = SparkSession
        .builder()
        .appName("burk")
        .master("local")
        .config("spark.sql.shuffle.partitions", 1)
        .getOrCreate()
      import spark.implicits._
      import org.apache.spark.sql.functions._

      //读取数据
      val burksDF: DataFrame = spark
        .read
        .format("csv")
        .option("sep", ",")
        .schema("burk STRING,year STRING,tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE,tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE,tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
        .load("data/burks.txt")

      burksDF.show()

      /**
       * 1、统计每个公司每年按月累计收入  行转列 --> sum窗口函数
       * 输出结果
       * 公司代码,年度,月份,当月收入,累计收入
       */

      /**
       * sql
       */
      burksDF.createOrReplaceTempView("burks")

      spark.sql(
        """
          |select
          |burk,year,month,plc,
          |sum(plc) over(partition by burk,year order by month) as leiji
          | from (
          |select burk,year,month,plc
          |from burks
          |lateral view explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) T as month,plc
          |) as a
          |
          |""".stripMargin)
      // .show(100)

      /**
       * DSL
       *
       */

      val m: Column = map(
        expr("1"), $"tsl01",
        expr("2"), $"tsl02",
        expr("3"), $"tsl03",
        expr("4"), $"tsl04",
        expr("5"), $"tsl05",
        expr("6"), $"tsl06",
        expr("7"), $"tsl07",
        expr("8"), $"tsl08",
        expr("9"), $"tsl09",
        expr("10"), $"tsl10",
        expr("11"), $"tsl11",
        expr("12"), $"tsl12"
      )

      burksDF
        //一行转换成多行
        .select($"burk", $"year", explode(m) as Array("month", "plc"))
        //计算按月累计
        .withColumn("leiji", sum($"plc") over Window.partitionBy($"burk", $"year").orderBy($"month"))
        .show(100)


    }
}
